package com.chinamobile.iot.mqtt.server.netty;

import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.mqtt.spi.security.ISslContextCreator;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.io.IOException;


@Component
public class NettyAcceptor {

    private static final Logger logger = LoggerFactory.getLogger(NettyAcceptor.class);

    /**
     * 在TCP连接成功后，收到Connect消息之前的超时时间间隔，默认为20秒
     */
    private static final int MAX_CONNECT_TIMEOUT = 60;

    @Autowired
    private NettyMQTTHandler handler;
    @Autowired
    private MqttConfiguration config;
    @Autowired
    private ISslContextCreator contextCreator;
    /**
     * 监听的事件循环
     */
    private EventLoopGroup m_bossGroup;
    /**
     * 工作的事件循环
     */
    private EventLoopGroup m_workerGroup;


    public void initialize() throws IOException {
        m_bossGroup = new NioEventLoopGroup(3);
        m_workerGroup = new NioEventLoopGroup(10);
        initializePlainTCPTransport();
        initializeSSLTCPTransport();
    }

    private void initFactory(String host, int port, final PipelineInitializer pipeliner) {
        ServerBootstrap b = new ServerBootstrap();
        b.group(m_bossGroup, m_workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        try {
                            pipeliner.init(pipeline);
                        } catch (Throwable th) {
                            logger.error("Severe error during pipeline creation", th);
                            throw th;
                        }
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .option(ChannelOption.SO_REUSEADDR, true)
                // 设置写Buffer的水位
                .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1024 * 8, 1024 * 16))
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.TCP_NODELAY, true);
        try {
            ChannelFuture f = b.bind(host, port);
            f.sync();
        } catch (InterruptedException ex) {
            logger.error(null, ex);
        }
    }

    private void initializePlainTCPTransport() throws IOException {
        final MqttIdleTimeoutHandler timeoutHandler = new MqttIdleTimeoutHandler();
        String host = config.getHost();
        int port = config.getPort();
        initFactory(host, port, new PipelineInitializer() {
            @Override
            void init(ChannelPipeline pipeline) {
                pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, MAX_CONNECT_TIMEOUT));
                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
                //pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
                pipeline.addLast("decoder", new MqttDecoder());
                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
                //pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
                pipeline.addLast("handler", handler);
            }
        });
        logger.info("Started TCP on host: {}, port {}", host, port);
    }

    private void initializeSSLTCPTransport() throws IOException {

        String host = config.getHost();
        int sslPort = config.getSsl().getPort();

        final MqttIdleTimeoutHandler timeoutHandler = new MqttIdleTimeoutHandler();
        final SSLContext sslContext = contextCreator.initSSLContext();
        if (sslContext == null){
            return ;
        }
        initFactory(host, sslPort, new PipelineInitializer() {
            @Override
            void init(ChannelPipeline pipeline) throws Exception {
                pipeline.addLast("ssl", createSslHandler(sslContext));
                pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, config.getConnectTimeout()));
                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
                pipeline.addLast("decoder", new MqttDecoder());
                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
                pipeline.addLast("handler", handler);
            }
        });
        logger.info("Started SSL on host: {}, port {}", host, sslPort);
    }

//    private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
//        logger.info("Configuring Websocket MQTT transport");
////        String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
////        if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
////            // Do nothing no WebSocket configured
////            logger.info("Property {} has been setted to {}. Websocket MQTT will be disabled",
////                    BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
////            return;
////        }
//        int port = 9883;
//
//        final MqttIdleTimeoutHandler timeoutHandler = new MqttIdleTimeoutHandler();
//
//        String host = "localhost";
//        initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
//
//            @Override
//            void init(ChannelPipeline pipeline) {
//                pipeline.addLast(new HttpServerCodec());
//                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
//                pipeline.addLast("webSocketHandler",
//                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
//                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
//                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
//                pipeline.addLast("decoder", new MqttDecoder());
//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
//                pipeline.addLast("handler", handler);
//            }
//        });
//    }

    public void close() {
        if (m_workerGroup == null) {
            throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
        }
        if (m_bossGroup == null) {
            throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
        }
        Future workerWaiter = m_workerGroup.shutdownGracefully();
        Future bossWaiter = m_bossGroup.shutdownGracefully();

        try {
            workerWaiter.await(100);
        } catch (InterruptedException iex) {
            throw new IllegalStateException(iex);
        }

        try {
            bossWaiter.await(100);
        } catch (InterruptedException iex) {
            throw new IllegalStateException(iex);
        }

//        MessageMetrics metrics = m_metricsCollector.computeMetrics();
//        logger.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote());

//        BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics();
//        logger.info(String.format("Bytes read: %d, bytes wrote: %d", bytesMetrics.readBytes(), bytesMetrics.wroteBytes()));
    }

    private ChannelHandler createSslHandler(SSLContext context) {
        SSLEngine sslEngine = context.createSSLEngine();
        sslEngine.setNeedClientAuth(false);
        sslEngine.setUseClientMode(false);
        sslEngine.setWantClientAuth(false);
        return new SslHandler(sslEngine);
    }

    abstract class PipelineInitializer {

        abstract void init(ChannelPipeline pipeline) throws Exception;
    }
}
